package bigdata.hourglass.engine.main


import akka.actor.ActorSystem
import bigdata.hourglass.common.utils.ZkUtils
import bigdata.hourglass.engine.{ExeActor, HourglassSession}
import bigdata.hourglass.engine.repl.SparkInterpreter
import bigdata.hourglass.utils.{AkkaRpcProviderFactory, ParamsUtil}
import org.apache.spark.LSQLConf.HOURGLASS_PARALLELISM
import org.apache.spark.{LSQLConf, SparkConf}

import scala.collection.JavaConverters._

/**
 * @author leo.jie (weixiao.me@aliyun.com)
 * @organization DataReal
 * @version 1.0
 * @website https://www.jlpyyf.com
 * @date 2019/12/12 12:01 上午
 * @since 1.0
 */
object HourglassEngineApp {
  var schedulerMode: Boolean = true

  def main(args: Array[String]): Unit = {
    val params = new ParamsUtil(args)
    require(params.hasParam("hourglass.name"), "Application name should be set")
    val tempParams = new java.util.HashMap[Any, Any]()
    params.getParamsMap.asScala.foreach { f => tempParams.put(f._1, f._2) }

    val interpreter = new SparkInterpreter(tempParams)
    val sparkConf: SparkConf = interpreter.start()

    val appPort = params.getIntParam(LSQLConf.L_AKKA_ENGINE_PORT.key, LSQLConf.L_AKKA_ENGINE_PORT.defaultValue.get)
    val akkaLogLevel = params.getParam(LSQLConf.L_AKKA_LOG_LEVEL.key, LSQLConf.L_AKKA_LOG_LEVEL.defaultValueString)
    val zkAddress = params.getParam(LSQLConf.L_SQL_ZK.key, LSQLConf.L_SQL_ZK.defaultValueString)
    val engineTag = params.getParam(LSQLConf.L_SQL_ENGINE_TAG.key, LSQLConf.L_SQL_ENGINE_TAG.defaultValueString)

    val akkaRpcProviderFactory: AkkaRpcProviderFactory = new AkkaRpcProviderFactory()
    akkaRpcProviderFactory.initConfig(appPort, akkaLogLevel)
    val actorConf = akkaRpcProviderFactory.getAkkaConfig(ZkUtils.getZkClient(zkAddress))
    val hostname = akkaRpcProviderFactory.getIp
    val port = akkaRpcProviderFactory.getPort
    val hourglassSession: HourglassSession = new HourglassSession(hostname + ":" + port, engineTag)
    val actorSystem: ActorSystem = ActorSystem("HourglassEngineSystem", actorConf)
    (1 to sparkConf.getInt(HOURGLASS_PARALLELISM.key, HOURGLASS_PARALLELISM.defaultValue.get))
      .foreach(id => actorSystem.actorOf(ExeActor.props(tempParams, interpreter, hourglassSession, sparkConf), name = s"actor$id"))
    hourglassSession.awaitTermination()
  }
}
